from datetime import timedelta, datetime
from airflow import DAG
from airflow.operators.python import PythonOperator


def print_task():
    print('这个sample_task2')
    try:
        return 1 / 0
    except Exception as e:
        print(f'出错了: {e}')


default_args = {
    # 用户
    'owner': 'jueyi',
    # 是否开启任务链依赖
    'depends_on_past': True,
    'email': ['1950698638@qq.com'],
    # 启动时间
    'start_date': datetime(2024, 8, 30),
    # 出错是否发邮件报警
    'email_on_failure': False,
    # 重试是否发邮件报警
    'email_on_retry': False,
    # 重试次数
    'retries': 1,
    # 重试时间间隔
    'retry_delay': timedelta(minutes=1),
}
# 声明任务图
dag = DAG(
    'airflow_sample_dag',
    default_args=default_args,
    description='airflow dag 样例',
    # 调度的时间
    schedule_interval=timedelta(days=1),
)

# 创建任务
sample_task1 = PythonOperator(
    # 任务id
    task_id='sample_task1',
    python_callable=lambda: print("hello world"),
    # 重试次数
    retries=3,
    # 把任务添加到任务图中
    dag=dag,
)

sample_task2 = PythonOperator(
    # 任务id
    task_id='sample_task2',
    python_callable=print_task,
    # 重试次数
    retries=3,
    # 把任务添加到任务图中
    dag=dag,
)

# 设置任务依赖
sample_task2.set_upstream(sample_task1)
